package com.education.realtime.app.dwd;

import com.education.realtime.app.BaseSQLApp;
import com.education.realtime.common.Constant;
import com.education.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdStudyApp extends BaseSQLApp {
    public static void main(String[] args) {

        new DwdStudyApp().init(2006,"DwdStudyApp");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
            tEnv.executeSql("create table dwd_traffic_vedio(\n" +
                    "play_sec string,\n" +
                    "position_sec string,\n" +
                    "video_id string,\n" +
                    "common map<string,string>,\n" +
                    "ts string\n" +
                    ")" + SqlUtil.getKafkaReadDDL(Constant.TOPIC_DWD_TRAFFIC_APPVIDEO,"DwdStudyApp"));

        Table table = tEnv.sqlQuery("select \n" +
                "`common`['uid'] uid," +
                "`common`['mid'] mid," +
                "play_sec ,\n" +
                "position_sec ,\n" +
                "video_id ,\n" +
                "ts " +
                "from dwd_traffic_vedio");

        tEnv.executeSql("create table dwd_study_learn(\n" +
                " uid string ,\n" +
                " mid string ,\n" +
                " play_sec string ,\n" +
                " position_sec string ,\n" +
                " video_id string, " +
                " ts string )" + SqlUtil.getKafkaWriteDDL(Constant.TOPIC_DWD_STUDY_LEARN));

        table.executeInsert("dwd_study_learn");
    }
}
